package com.wx.task.future.fj;

import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.URL;
import java.util.*;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 利用 forkJoin 线程池
 * 完成文本的单词统计
 * MapReduce
 *
 * @author wang.xun
 * @date 2023/08/04
 */
@Slf4j
public class WordCount {


    final static List words = new ArrayList<String>();

    public static void main(String[] args) {
        ForkJoinPool forkJoinPool = new ForkJoinPool(5);
        URL resource = WordCount.class.getClassLoader().getResource("wc.txt");
        WordCount.readData(resource.getFile());
        Map<String, Integer> invoke = forkJoinPool.invoke(new MR(words, 0, words.size()));
        invoke.forEach((k,v)->{
            log.info(" key : 【{}】 count :【{}】", k, v);
        });

    }

    /**
     * 读取数据
     *
     * @param fileUrl 文件url
     */
    static void readData(String fileUrl) {
        try {
            FileReader fileReader = new FileReader(new File(fileUrl));
            BufferedReader bufferedReader = new BufferedReader(fileReader);
            String line = "";
            while ((line = bufferedReader.readLine()) != null) {
                words.add(line);
            }

        } catch (FileNotFoundException e) {
            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    static class MR extends RecursiveTask<Map<String, Integer>> {
        List<String> counts;
        int startIndex;
        int endIndex;

        public MR(List counts, int startIndex, int endIndex) {
            this.counts = counts;
            this.startIndex = startIndex;
            this.endIndex = endIndex;
        }

        @Override
        protected Map<String, Integer> compute() {
            if(endIndex - startIndex == 1){
                // 获取一行数据
                return statistics(counts.get(startIndex));
            }else {
                int midIndex = (startIndex+endIndex)/2;
                MR mr1 = new MR(counts, startIndex, midIndex);
                mr1.fork();
                MR mr2 = new MR(counts, midIndex, endIndex);


               return merge(mr1.join(), mr2.compute());

            }
        }

        Map<String, Integer> statistics(String line) {
            Map<String, Integer> result =  new HashMap<>();
            String[] strings = line.split("\\s+");
            Stream.of(strings).forEach((s) -> {
                if (result.containsKey(s)) {
                    result.put(s, result.get(s) + 1);
                }else {
                    result.put(s,  1);
                }
            });
            return result;
        }


        /**
         * 合并
         *
         * @param map1 map1
         * @param map2 map2
         * @return {@link Map}<{@link String}, {@link Integer}>
         */
        private Map<String, Integer> merge(Map<String, Integer> map1,Map<String, Integer> map2){

            HashMap<String, Integer> rs = new HashMap<>();
            rs.putAll(map1);

            map2.forEach((k,v)->{
                Integer value = map1.get(k);
                if(value != null){
                    v += value;
                }
                rs.put(k, v);
            });
            return rs;
        }
    }
}
